package com.sui.bigdata.rtcadmin.service.impl;

import com.alibaba.fastjson.JSON;
import com.sui.bigdata.flink.table.client.JobClient;
import com.sui.bigdata.flink.table.client.SqlClient;
import com.sui.bigdata.rtcadmin.dto.MetaDto;
import com.sui.bigdata.rtcadmin.exception.*;
import com.sui.bigdata.rtcadmin.repository.mapper.MetaMapper;
import com.sui.bigdata.rtcadmin.repository.model.CreateInfo;
import com.sui.bigdata.rtcadmin.repository.model.Meta;
import com.sui.bigdata.rtcadmin.service.MetaService;
import com.sui.bigdata.rtcadmin.util.BasicPropUtils;
import com.sui.bigdata.rtcadmin.util.CreateTableParser;
import com.sui.bigdata.rtcadmin.util.MetaUtils;
import com.sui.bigdata.rtcadmin.util.ScanUdfUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;

/**
 * @author YongChen
 * @date 2019/11/6 14:41
 * @description
 * @email yong_chen@sui.com
 */
@Service
public class MetaServiceImpl implements MetaService {
    private static Logger logger = LoggerFactory.getLogger(MetaServiceImpl.class);

    @Autowired
    private MetaMapper metaMapper;

    @Autowired
    private MetaUtils metaUtils;

    @Value("${flink.udf.jar.path}")
    private String flinkUdfJarPath;

    @Override
    public void insert(CreateInfo createInfo) throws Exception{

        if (StringUtils.isBlank(createInfo.getCreateSql()) || StringUtils.isBlank(createInfo.getDatabaseName())){
            throw new CreateSqlBlankException();
        }
        String databaseName = createInfo.getDatabaseName();
        StringBuffer repetTable = new StringBuffer();
        String newSql = createInfo.getCreateSql();
        if(!"DEFAULT".equals(databaseName.toUpperCase())){
            newSql = "USE " + databaseName +";"+createInfo.getCreateSql();
        }
        Boolean flag = false;
        try{
            flag = SqlClient.ddl(UUID.randomUUID().toString(), newSql);
        }catch (TableAlreadyExistException var1){
            logger.warn("表已经存在！",var1);
            throw var1;
        }catch (Exception var2){
            logger.error("表创建异常！", var2);
            throw var2;
        }
        if (flag){
            List<CreateTableParser.SqlParserResult> result = CreateTableParser.newInstance().parseSql(createInfo.getCreateSql());
            for(CreateTableParser.SqlParserResult sqlParserResult : result){
                Meta meta = new Meta();
                meta.setTableName(sqlParserResult.getTableName());
                if (metaMapper.queryByTableNameAndDB(sqlParserResult.getTableName(), databaseName) != null){
                    repetTable.append(sqlParserResult.getTableName());
                    repetTable.append(" ");
                    continue;
                }
                Object component = sqlParserResult.getPropMap().get("'connector'");
                if(component != null){
                    meta.setComponent(component.toString());
                }
                String[] fieldsInfoStr = sqlParserResult.getFieldsInfoStr().split(",");
                String formateFields = Arrays.stream(fieldsInfoStr).map(String::trim).collect(Collectors.joining(","));
                meta.setFieldInfo(formateFields);
                meta.setCreateTableStatement(sqlParserResult.getCreateTableStatement());
                meta.setProp(JSON.toJSONString(sqlParserResult.getPropMap()));
                meta.setDatabaseName(databaseName);
                metaMapper.insert(meta);
            }
        }
        if (StringUtils.isNotBlank(repetTable.toString())){
            throw new TableNameRepeteException(repetTable.toString());
        }
    }
    @Override
    public void delete (String tableName, String databaseName) throws Exception{
        if (StringUtils.isBlank(tableName) || StringUtils.isBlank(databaseName)){
            throw new CreateSqlBlankException();
        }
        String sql = "drop table IF EXISTS "+ tableName +";";
        if(!"DEFAULT".equals(databaseName.toUpperCase())){
            sql = "drop table IF EXISTS " + databaseName + "." + tableName + ";";
        }
        Boolean flag = SqlClient.ddl(UUID.randomUUID().toString(), sql);
        if(flag){
            metaMapper.delete(tableName, databaseName);
        }
    }


    @Override
    public List<MetaDto> queryAll() {
        List<Meta> metas = metaMapper.queryAll();
        return metas.stream().map(MetaDto :: buildMetaDto).collect(Collectors.toList());
    }


    @Override
    public List<MetaDto> queryByType(String type,String component) {
        List<Meta> metas = metaMapper.queryByType(type,component);
        return metas.stream().map(MetaDto :: buildMetaDto).collect(Collectors.toList());
    }

    @Override
    public MetaDto queryByTableName(String tableName) {
        Meta meta =  metaMapper.queryByTableName(tableName);
        if ( meta == null){
            throw new TableNotFoundException();
        }
        return MetaDto.buildMetaDto(meta);
    }

    @Override
    public String list(String type) {
        return metaUtils.listComponent(type);
    }

    @Override
    public Object functionOperation(String action) throws Exception {
        try{
            if("flush".equals(action)){
                Boolean udfRegest = SqlClient.ddl(UUID.randomUUID().toString(), ScanUdfUtils.getUdfs(flinkUdfJarPath));
                return  udfRegest ? "操作成功！" :  "操作失败！";
            }else if("show".equals(action)){
                List<String> list = SqlClient.show(UUID.randomUUID().toString(), "show functions;");
                return  list;
            }else{
                return new ActionNotSupportException();
            }

        }catch (Exception e){
            logger.error("udf 操作异常！", e);
            throw e;
        }
    }
}
